package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.UniqueVisitStats;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple4;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class UniqueVisit10sApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //
//        env.enableCheckpointing(5 * 60000L);
//        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
//        //
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/210927/flink-state");


        // 从kafka获取数据 dwd_page_log
        String sourceTopic = "dwd_page_log";
        String groupId = "unique_visit";

        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));

        // 将数据转换为对应jsonObj 指定watermark
        SingleOutputStreamOperator<JSONObject> jsonObjWithWM = kafkaDS.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }));

        // 按照mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWM.keyBy(key -> key.getJSONObject("common").getString("mid"));

        // 使用状态编程的方式按照mid去重
        SingleOutputStreamOperator<JSONObject> filterDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            SimpleDateFormat sdf = null;
            ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                sdf = new SimpleDateFormat("yyyy-MM-dd");

                // 设置状态的ttl
                ValueStateDescriptor<String> state = new ValueStateDescriptor<>("state", String.class);
                state.enableTimeToLive(new StateTtlConfig
                        .Builder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                valueState = getRuntimeContext().getState(state);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {

                // 获取上一个页面id
                String lastPage = value.getJSONObject("page").getString("last_page_id");
                // 获取访问时间
                Long ts = value.getLong("ts");
                String firstVisit = sdf.format(ts);
                // 获取状态
                String firstVisitState = valueState.value();

                // 如果上一个页面id为空 且 (状态为空 或 状态的值和当前值不同) 则为当日第一次登录
                if (lastPage == null && (firstVisitState == null || !firstVisit.equals(firstVisitState))) {
                    valueState.update(firstVisit);
                    return true;
                }

                return false;
            }
        });
        filterDS.print("filterDS>>>>>");

        // 转换为JavaBean对象 4个维度
        SingleOutputStreamOperator<UniqueVisitStats> uniqueVisitStatsDS = filterDS.map(jsoObj -> {
            JSONObject common = jsoObj.getJSONObject("common");
            return new UniqueVisitStats("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L,
                    jsoObj.getLong("ts"));
        });

        // 指定key
        KeyedStream<UniqueVisitStats, Tuple4<String, String, String, String>> uniqueVisitStatsTuple4KeyedStream = uniqueVisitStatsDS.keyBy(new KeySelector<UniqueVisitStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(UniqueVisitStats value) throws Exception {
                return new Tuple4<>(value.getAr(),
                        value.getCh(),
                        value.getIs_new(),
                        value.getVc());
            }
        });

        // 开10s的滚动窗口
        WindowedStream<UniqueVisitStats, Tuple4<String, String, String, String>, TimeWindow> window = uniqueVisitStatsTuple4KeyedStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        // 聚合
        SingleOutputStreamOperator<UniqueVisitStats> resultDS = window.reduce(
                new ReduceFunction<UniqueVisitStats>() {
                    @Override
                    public UniqueVisitStats reduce(UniqueVisitStats value1, UniqueVisitStats value2) throws Exception {
                        value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                        return value1;
                    }
                },
                new WindowFunction<UniqueVisitStats, UniqueVisitStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<UniqueVisitStats> input, Collector<UniqueVisitStats> out) throws Exception {
                        UniqueVisitStats uniqueVisitStats = input.iterator().next();

                        // 补充窗口信息
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        uniqueVisitStats.setStt(sdf.format(window.getStart()));
                        uniqueVisitStats.setEdt(sdf.format(window.getEnd()));

                        // 输出数据
                        out.collect(uniqueVisitStats);
                    }
                });

        resultDS.print("resultDS>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_uv_vc_ch_isnew_ar_10s_2022 values(?,?,?,?,?,?,?,?)"));


        env.execute("UniqueVisit10sApp");
    }
}
